[Amazon SageMaker] iPhoneをRTSPサーバにしてイメージ分類(Image Classification)を試してみました

SIN

2020.05.21

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX事業本部の平内（SIN）です。

前回、Webカメラでイメージ分類を行う場合、解像度は、あまり関係なく「距離」に限界があるという事になってしまいました。

って事で、試しにカメラを変えてみました。

今回試したのは、iPhone5です。 iPhoneをRTSPサーバにして、カメラの画像をリアルタイムに取得し、イメージ分類してみました。使用したモデルは前回と同じで、17種類のお菓子を分類するものです。

カメラと商品の距離は、5ｍです。

2 RTSPサーバ(iPhone)

iPhoneをRTSPサーバにするために、下記のアプリを利用させて頂きました。インストールするだけで簡単に利用可能です。

Live-Reporter スマートフォンをライブカメラに

設定は、以下のとおりとしました。

3 コード

確認に使用したコードは、以下のとおりです。 OpenCVのVideoCapture()は、パラメータにURLを指定することで、リモートサーバを入力とする事ができます。

<br />import json
import datetime
import cv2
from boto3.session import Session

PROFILE = 'developer'
END_POINT = 'sampleEndPoint'
CLASSES = ['PORIPPY(GREEN)', 'OREO', 'CUNTRY_MAM', 'PORIPPY(RED)', 'BANANA'
        , 'CHEDDER_CHEESE', 'PRETZEL(YELLOW)', 'FURUGURA(BROWN)', 'NOIR'
        , 'PRIME', 'CRATZ(RED)', 'CRATZ(GREEN)', 'PRETZEL(BLACK)', 'CRATZ(ORANGE)'
        , 'ASPARA', 'FURUGURA(RED)', 'PRETZEL(GREEN)']

RTSP = 'rtsp://192.168.0.113/'

TARGET_WIDTH = 200
TARGET_HEIGHT = int(TARGET_WIDTH*1.5)
BIAS = 100
RESIZE = 0.5
FONT_SIZE = 1

class SageMaker():
    def __init__(self, profile, endPoint):
        self.__end_point = endPoint
        self.__client = Session(profile_name=profile).client('sagemaker-runtime')

    def invoke(self, image):
        data = self.__client.invoke_endpoint(
            EndpointName=self.__end_point,
            Body=image,
            ContentType='image/jpeg'
        )
        return json.loads(data['Body'].read())

def putText(frame, x, y, font_size, text):
    font_color = (255, 255, 255)
    cv2.putText(frame, text, (x, y), cv2.FONT_HERSHEY_SIMPLEX, font_size, font_color, 2, cv2.LINE_AA)

def onClick(event, x, y, flags, frame):
    if event == cv2.EVENT_LBUTTONUP:
        now = datetime.datetime.now()
        cv2.imwrite(str(now) + '.jpg', frame)
        print("Saved.")

def createArea(width, height, w, h, bias):
    x_1 = int(width/2-w/2)
    x_2 = int(width/2+w/2)
    y_1 = int(height/2-h/2) - bias
    y_2 = int(height/2+h/2) - bias
    return [x_1, y_1, x_2, y_2]

def main():

    cap = cv2.VideoCapture(RTSP)

    fps = cap.get(cv2.CAP_PROP_FPS)
    width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
    height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
    print("FPS:{}　WIDTH:{}　HEIGHT:{}".format(fps, width, height))

    area = createArea(width, height, TARGET_WIDTH, TARGET_HEIGHT, BIAS)

    sageMake = SageMaker(PROFILE, END_POINT)

    while True:

        # カメラ画像取得
        _, frame = cap.read()
        if(frame is None):
            continue

        img = frame[area[1]: area[3], area[0]: area[2]]

        _, jpg = cv2.imencode('.jpg', img)
        results = sageMake.invoke(jpg.tostring())
        probabilitys = {}
        for i, result in enumerate(results):
            probabilitys[CLASSES[i]] = result

        probabilitys = sorted(probabilitys.items(), key = lambda x:x[1], reverse=True)
        for i in range(3):
            (name, probability) = probabilitys[i]
            text = "{} {}".format(name, probability)
            putText(frame, 20, int(height) - 120 + i * (FONT_SIZE * 50), FONT_SIZE, text)

        # 対象範囲の枠表示
        frame = cv2.rectangle(frame, (area[0], area[1]), (area[2], area[3]), (255, 255, 0), 3)

        # フレーム表示
        frame = cv2.resize(frame, (int(width*RESIZE), int(height*RESIZE)))
        cv2.imshow('frame', frame)

        # マウスクリックでスナップ撮影
        cv2.setMouseCallback('frame', onClick, frame)

        cv2.waitKey(1)

        cap.release()
        cv2.destroyAllWindows()

main()